package com.chiaic.unittest.linktest;

import com.alibaba.fastjson2.JSON;
import com.chiaic.protobuf.proto.TestProto;
import com.chiaic.unittest.common.thread.UnitTestThreadFactory;
import com.chiaic.unittest.data.MessagePackage;
import com.chiaic.unittest.data.ProtoType;
import com.chiaic.unittest.data.TsData;
import com.chiaic.unittest.unit.KafkaQueue;
import com.chiaic.unittest.unit.countService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;

import java.util.Random;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;

import static java.lang.System.currentTimeMillis;
import static java.util.concurrent.TimeUnit.SECONDS;

//@Service
@Slf4j
//@Component
public class LinkTestService {

    //    private final ConcurrentLinkedQueue<MessagePackage> queue = new ConcurrentLinkedQueue<>();
//        private final LinkedBlockingQueue<MessagePackage> queue = new LinkedBlockingQueue<>();
    private final KafkaQueue<MessagePackage> queue;
    private final ExecutorService dispatcherInExecutor;

    private final ExecutorService dispatcherOutExecutor;

    private final countService countService;

    private final String IN = "queueIn";
    private final String OUT = "queueOut";

    private final String SIZE = "queueSize";

    private final String TO = "queueTimeout";

    private long batchSize = 100000;

    private long timeOut = 1000;

    private int sleepTime = 5;

    private ProtoType protoType = ProtoType.PROTO_BUFF;


    //    @Autowired
    public LinkTestService(countService countService,
                           @Value("${linktest.dispatcherOutThreads:1}") int dispatcherOutThreads,
                           @Value("${linktest.dispatcherInThreads:1}") int dispatcherInThreads,
                           @Value("${linktest.protoType:PROTO_BUFF}") ProtoType protoType) {
        this.countService = countService;
        this.protoType = protoType;

        queue = new KafkaQueue<MessagePackage>("172.16.32.212:9092", 1);

        countService.initCounter(OUT, true);
        countService.initCounter(IN, true);
        countService.initCounter(SIZE);
        countService.initCounter(TO);

        if (dispatcherInThreads > 0) {
            this.dispatcherInExecutor = new ThreadPoolExecutor(dispatcherInThreads,
                    dispatcherInThreads, 60, SECONDS,
                    new LinkedBlockingQueue<>(), UnitTestThreadFactory.forName("queueTest-In-"));
            for (int i = 0; i < dispatcherInThreads; i++) {

                dispatcherInExecutor.submit(this::dispatchIn);
            }
        } else {
            this.dispatcherInExecutor = null;
        }

        if (dispatcherOutThreads > 0) {
            this.dispatcherOutExecutor = new ThreadPoolExecutor(dispatcherOutThreads,
                    dispatcherOutThreads, 60, SECONDS,
                    new LinkedBlockingQueue<>(), UnitTestThreadFactory.forName("queueTest-Out-"));


            for (int i = 0; i < dispatcherOutThreads; i++) {

                dispatcherOutExecutor.submit(this::dispatchOut);
            }
        } else {
            this.dispatcherOutExecutor = null;
        }
    }

    private MessagePackage decodeTsData(MessagePackage src) {
        switch (protoType) {
            case PROTO_BUFF:
                break;
            case JSON:
                src.setSrcData(JSON.parseObject(src.getJsonData(), TsData.class));
                break;
        }
        return src;
    }

    private void dispatchOut() {
        while (true) {

            MessagePackage messagePackage = queue.poll();
            if (messagePackage == null) {
                try {
                    Thread.sleep(50);
                    continue;
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
            messagePackage = decodeTsData(messagePackage);
            if (messagePackage.getSrcData().getServerTs() + timeOut < currentTimeMillis()) {
                countService.add(TO);
            }

            countService.add(OUT);
            countService.addCount(SIZE, -1);

            while ((messagePackage = queue.poll()) != null) {
                messagePackage = decodeTsData(messagePackage);
                if (messagePackage.getSrcData().getServerTs() + timeOut < currentTimeMillis()) {
                    countService.add(TO);
                }

                countService.add(OUT);
                countService.addCount(SIZE, -1);
            }


        }
    }

    private void inqueue(Random random) {
        for (int i = 0; i < batchSize; i++) {
            TsData data = TsData.builder().serverTs(currentTimeMillis())
                    .uuid(UUID.randomUUID().toString())
                    .dataTs(currentTimeMillis())
                    .doubleValue(random.nextDouble())
                    .build();


            MessagePackage messagePackage = null;

            switch (protoType) {
                case PROTO_BUFF:
                    TestProto.TsDataProtoBuffMessage.Builder protoBuilder = TestProto.TsDataProtoBuffMessage.newBuilder();
                    protoBuilder.setServerTs(data.getServerTs())
                            .setUuid(data.getUuid())
                            .setDoubleValue(data.getDoubleValue()).build();
                    TestProto.TsDataProtoBuffMessage ms = protoBuilder.build();

                    messagePackage = MessagePackage.builder().srcData(data).protoData(ms.toByteArray()).build();
                    break;
                case JSON:
                    messagePackage = MessagePackage.builder().srcData(data).JsonData(JSON.toJSONString(data)).build();
                    break;
            }


            queue.add(messagePackage);
            countService.add(IN);
            countService.add(SIZE);
        }
    }

    private void dispatchIn() {
        Random random = new Random(currentTimeMillis());
        while (true) {

            inqueue(random);


//            try {
//                Thread.sleep(sleepTime);
//            } catch (InterruptedException e) {
//                throw new RuntimeException(e);
//            }
        }
    }
}
